package com.qst.mapreduce.wordcount.shiyanwu;


import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;

public class JoinMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    HashMap<String, String> nameMap=new HashMap();

    @Override
    protected void setup(Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        //通过缓存文件得到姓名文件地址
        URI[] cacheFile=context.getCacheFiles();
        Path path=new Path(cacheFile[0]);
        //获取文件系统对象，并开流
        FileSystem fs= FileSystem.get(context.getConfiguration());
        FSDataInputStream fis=fs.open(path);
        //通过包装流转换为reader，方便按行读取
        BufferedReader reader=new BufferedReader(new InputStreamReader(fis, "UTF-8"));
        //逐行读取，按行处理
        String line;
        while(StringUtils.isNotEmpty(line= reader.readLine())){
            String[] split=line.split(",");
            nameMap.put(split[0],split[1]);
        }
        //关闭流
        IOUtils.closeStream(reader);
    }
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] words = line.split(" ");

        String id=words[6];
        String name=nameMap.get(id);
        //Text k = new Text(words[6]);
        //IntWritable V=new IntWritable(1);
        //IntWritable V=new IntWritable(Integer.parseInt(words[7]));
        IntWritable V=new IntWritable(1);
        context.write(new Text(name),V);


    }
}
